%%% -------------------------------------------------------------------
%%% Author  : mjollnir.ray
%%% Description :
%%%
%%% Created : 2009-3-9
%%% -------------------------------------------------------------------
-module(bench_server).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include("bench.hrl").
-include("map.hrl").
-include("ring.hrl").

%% --------------------------------------------------------------------
%% External exports
-export([start_link/0]).
-export([send_testdata/1, collect_experiment_data/1, stop_collect/0]).
-export([begin_map_rangers/2, end_map_rangers/0, report_state/0]).
-export([pressure_test/1,stop_pressure_test/0, write_measure/1]).

-export([nodes_parallel_test/1, nodes_parallel_test/0]).
-export([remote_db_test/0, remote_db_test/1, local_db_test/0,local_db_test/1,list_keys/0]).
-export([plocal_db_test/1, plocal_db_test/2, premote_db_test/1, premote_db_test/2, parallel_remote_db_test/2]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-define(SERVER, ?MODULE).

-record(state, {keys_for_test=[], db_io=#db_io_testdata{}, map_rangers=[], collect_period = 0, collect_num = 0, collect_timer}).

%% ====================================================================
%% External functions
%% ====================================================================
start_link() ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

send_testdata(#db_io_testdata{} = TestData) ->
    gen_server:cast(?MODULE, TestData).

%% ====================================================================
%% Server functions
%% ====================================================================
nodes_parallel_test() ->
    nodes_parallel_read(2000).

nodes_parallel_read(N) when N<5000 ->
    io:format("N: ~p~n", [N]),
    nodes_parallel_test(N),
    io:format("sleep time: ~p~n", [time()]),
    timer:sleep(4*60*1000),
    report_state(),
    nodes_parallel_read(N+50);
nodes_parallel_read(_N) ->
    io:format("over!~n"),
    ok.

nodes_parallel_test(ParallelProcessCount) ->
    Nodes = lists:map(fun(#bench_node{name=Node}) -> Node end, ring_lord:get_benchnodes()),
    io:format("nodes: ~p~n", [Nodes]),
    gen_server:abcast(Nodes, ?SERVER, {start_parallel_remote_read_test, ParallelProcessCount}).

begin_map_rangers(Level, Num) ->
    gen_server:call(?SERVER, {add_map_ranger, Level, Num}).

end_map_rangers() ->
    gen_server:cast(?SERVER, {stop_map_ranger}).

report_state() ->
    gen_server:call(?SERVER, {report_state}).

write_measure(Level) ->
     {Time, Count} = timer:tc(data_importer,scan,[Level]),
     io:format("write to node(s) time: ~p(ms)~n", [Time/1000]),
     io:format("average time: ~.2f(ms)~n", [Time/(1000*Count)]).

pressure_test(RangerNum) ->
    begin_map_rangers(16, RangerNum),       % Level 18
    collect_experiment_data(60), %10 * 60),            % 10 * 60 sec = 10 min 
    TestDuration = 60 * 2,                      % test duration: 14 * 60 min = (14 hours)
    Time = (1000 * 60) * (TestDuration)+10, 
    timer:apply_after(Time, ?MODULE, stop_pressure_test, []).
    
stop_pressure_test() ->
    io:format("~p: STOP pressure test.~n", [now()]),
    stop_collect(),
    end_map_rangers().

collect_experiment_data(IntervalTime) -> % milliseconds
%    gen_server:cast(?SERVER, {collect_data}).
    gen_server:cast(?SERVER, {start_collect, IntervalTime}).
%    {ok, TRef} = 
%        timer:apply_interval(IntervalTime*1000, gen_server, cast, [?SERVER, {collect_data, IntervalTime}]).

stop_collect() ->
    gen_server:cast(?SERVER, {stop_collect}).

send_period_data(PeriodTd) ->
    gen_server:cast(?SERVER, {period_data, PeriodTd}).

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init([]) ->
    process_flag(trap_exit, true),
    io:format("registED: ~p~n", [global:registered_names()]),
    ok = ring_lord:add_benchnode(node()),
    {ok, #state{}}.

%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call({report_state}, _From, #state{db_io=DB, map_rangers=MapRangers} = State) ->
    AliveNum = lists:foldl(fun(Ranger, Sum) ->
                    IsAlive = is_process_alive(Ranger),
                    if 
                        IsAlive ->
                            Sum + 1;
                        true ->
                            Sum
                    end
                end, 0, MapRangers),
    Reply = [{db_io, DB}, 
             {total_ranger, length(MapRangers)}, 
             {alive_ranger, AliveNum}, 
             {process_count, erlang:system_info(process_count)}, 
             {memory,erlang:memory()}, 
             {bench_server, self()}],
    {reply, Reply, State};
handle_call({add_map_ranger, Level, Num}, _From, State) ->
    List = lists:seq(1, Num),
    MapRangers = [map_ranger:start(Level) || _No <- List],
    NewMapRangers = State#state.map_rangers ++ MapRangers,
    Reply = length(NewMapRangers),
    {reply, Reply, State#state{map_rangers = NewMapRangers}}.
    
%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast({start_parallel_remote_read_test, ParallelProcessCount}, #state{keys_for_test=TestKeys}=State) ->
    if 
        length(TestKeys) == 0 ->          
            Keys = list_keys();
        true ->
            Keys = TestKeys
    end,
    spawn(fun() -> premote_db_test(Keys, ParallelProcessCount) end),
    {noreply, State#state{keys_for_test=Keys}};
handle_cast({stop_map_ranger}, #state{map_rangers=MapRangers} = State) ->
    [MapRanger ! stop || MapRanger <- MapRangers, is_process_alive(MapRanger)],
%    lists:foreach(fun(MapRanger) -> MapRanger ! stop end, MapRangers),
    {noreply, State#state{map_rangers=[]}};
handle_cast({collect_data, CollectPeriod}, #state{map_rangers=MapRangers}=State) ->
    io:format("starT collect data, bench_server=~p~n", [self()]),
    spawn_link(fun() -> collect_period_data(MapRangers) end),
    {noreply, State#state{collect_period = CollectPeriod}};
handle_cast({period_data, PeriodTd}, State) ->
    PeriodCount = PeriodTd#db_io_testdata.count,
    Period = State#state.collect_period,
    if 
        (Period > 0) and (PeriodCount > 0) -> 
            CollectNum = State#state.collect_num,
            Runtime = PeriodTd#db_io_testdata.runtime,
            DataSize = PeriodTd#db_io_testdata.data_size,
            RangerNum = length(State#state.map_rangers),
            log_testdata(RangerNum, CollectNum, Period, PeriodCount, Runtime, DataSize),
            {noreply, State#state{collect_num = CollectNum + 1, db_io = #db_io_testdata{}}};
        true ->
            skip,
            {noreply, State}
    end;
handle_cast({start_collect, IntervalTime}, #state{collect_timer=OldTimer} = State) -> % milliseconds
    io:format("bench_server: ~p~n", [self()]),
    if 
        OldTimer =/= undefined ->
            timer:cancel(OldTimer);
        true ->
            ok
    end,
    {ok, TRef} = 
        timer:apply_interval(IntervalTime*1000, gen_server, cast, [?SERVER, {collect_data, IntervalTime}]),
    {noreply, State#state{collect_timer=TRef}};
handle_cast({stop_collect}, #state{collect_timer=CollectTimer} = State) -> % milliseconds
    if
        CollectTimer =/= undefined ->
            timer:cancel(CollectTimer);
        true ->
            ok
    end,
    {noreply, State#state{collect_num=0, collect_period=0, collect_timer=undefined}}.



log_testdata(RangerNum, CollectNum, Period, PeriodCount, Runtime, DataSize) ->
    {ok, IoDevice} = file:open("./bench.txt", [append]),
    %% time(sec)  hit_num/second   ms/hit_num  data_size/second  ranger_num, process_count
    io:format(IoDevice, "~p\t~p\t~.2f\t~.1f\t~p\t~p\r\n", 
              [CollectNum*Period, 
               erlang:trunc(0.5+PeriodCount/Period), 
               Runtime/PeriodCount, 
               DataSize/Period,
              RangerNum, erlang:system_info(process_count)]),
    file:close(IoDevice).
    
collect_period_data(MapRangers) ->
    Collector = self(),
    Invoker = fun() ->
                      lists:foreach(fun(MapRanger) -> MapRanger ! {collect_data, Collector} end, MapRangers)
              end,
    spawn(Invoker),
    PeriodTd = collect_replies(length(MapRangers), #db_io_testdata{}),
    send_period_data(PeriodTd).

collect_replies(0, AccTd) ->
    io:format("collect replies SUCCESSED!~n", []),
    AccTd;
collect_replies(N, AccTd) ->
    receive
        #db_io_testdata{} = TD ->
            collect_replies(N-1, sum_testdata(AccTd, TD));
        Any ->
            io:format("SOMETHING must be WRONG: ~p~n", [Any])
    after 1000 ->
            io:format("too BUSY for COLLECT replies(rest ~p)! timeout 1000ms~n", [N])
    end.
        

sum_testdata(#db_io_testdata{count=C1,runtime=R1,data_size=S1}, #db_io_testdata{count=C2,runtime=R2,data_size=S2}) ->
    #db_io_testdata{count=C1+C2, runtime=R1+R2, data_size=S1+S2}.
    
%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info({'EXIT', Pid, normal}, State) ->
    io:format("~p -> PROCESS: ~p died NORMAL: process_count=~p~n", [time(), Pid, erlang:system_info(process_count)]),
    io:format("bench_server: ~p~n", [self()]),
    {noreply, State};
handle_info({'EXIT', Pid, {system_limit, _CallStack}}, #state{map_rangers=MapRangers}=State) ->
%    io:format("process: ~p(is_alive=~p) died BY system_limit. process_count=~p~n", [Pid, is_process_alive(Pid), erlang:system_info(process_count)]),
%    io:format("bench_server: ~p~n", [self()]),
    NewMapRangers = [MapRanger || MapRanger <- MapRangers, MapRanger =/= Pid],
    {noreply, State#state{map_rangers=NewMapRangers}};
handle_info({'EXIT', Pid, Why}, State) ->
    io:format("Process: ~p(is_alive=~p) unnormal died with: ~p~nprocess_count=~p~n", [Pid, is_process_alive(Pid), Why, erlang:system_info(process_count)]),
    {noreply, State};
handle_info(Info, State) ->
    io:format("sOMEthing haPPened: ~p~n", [Info]),
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------
remote_db_test() ->
    Keys = list_keys(),
    remote_db_test(Keys).
remote_db_test(Keys) ->
    B = now(),
    {SumTime, Count} = 
    lists:foldr(fun(Key, {Sum,Count}) ->
                    {Time, #cell{}=_Cell} = 
                        timer:tc(ring_accessor, get_from_ring, [Key]),
%                    io:format("remote get: ~p~n", [_Cell]),
                    {Sum + Time, Count+1}
                end, 
                {0,0},
                Keys),
    io:format("TOTAL elapsed time: ~.3f(seconds)~n", [timer:now_diff(now(), B)/1000000]),
    io:format("average: ~.3f(ms)~n", [SumTime/(1000*length(Keys))]),
    io:format("count: ~p~n", [Count]).
 
parallel_remote_db_test(Start, Step) ->
    Keys = list_keys(),
    parallel_remote_db_test(Keys, Start, Step).

parallel_remote_db_test(Keys, Start, Step) when (Start < 4000) ->
    ParallelProcessCount = Start + Step,
    premote_db_test(Keys, ParallelProcessCount),
    node_readers:report_state(),
    timer:sleep(random:uniform(10) * 1000),
    parallel_remote_db_test(Keys, ParallelProcessCount, Step);
parallel_remote_db_test(_Keys, Start, _Step) ->
    io:format("STOP here: ~p~n(parallel process)~n", [Start]).
    
premote_db_test(N) ->
    Keys = list_keys(),
    premote_db_test(Keys, N).
premote_db_test(Keys, N) ->
    S = now(),
    io:format("S: ~p, parallel: ~p~n", [time(), N]),
    {TotalTime, TotalCount} = paral_remote_read(N, Keys, {0,0}),
    Elapse = timer:now_diff(now(), S),
    io:format("E: ~p, TOTAL elapse: ~.1f(seconds)~n", [time(), Elapse/1000000]),
    io:format("every request: ~.3f(ms)~n", [Elapse/(1000*TotalCount)]),
    {ok, IoDevice} = file:open("premote_bench.txt", [append]),
    %% Paralle num, Actual handle time, Average response time 
    io:format(IoDevice, "~p\t~.2f\t~.3f\r\n", [N, TotalTime/(1000*TotalCount), Elapse/(1000*TotalCount)]),
    file:close(IoDevice).
    

paral_remote_read(N, Keys, {AccCount, AccTime}) when N =< length(Keys) ->
    Collector = self(),
    {Keys2, Rest} = lists:split(N, Keys),
    B = now(),
    lists:foreach(
    fun(Key) ->
        spawn(fun() ->
            {Time, #cell{}=_Cell} = timer:tc(ring_accessor, get_from_ring, [Key]),
            Collector ! {tc, Time}
        end)
    end, Keys2),
    Count = length(Keys2),
    SumTime = collect_result(Count, 0),
    paral_remote_read(N, Rest, {AccCount+Count, AccTime+SumTime});
paral_remote_read(_N, Keys, {AccCount, AccTime})->
    Collector = self(),
    lists:foreach(
    fun(Key) ->
        spawn(fun() ->
            {Time, #cell{}=_Cell} = timer:tc(ring_accessor, get_from_ring, [Key]),
            Collector ! {tc, Time}
        end)
    end, Keys),
    Count = length(Keys),
    SumTime = collect_result(Count, 0),
    TotalCount = AccCount + Count,
    TotalTime = AccTime + SumTime,
    io:format("EACH request handle: ~.3f(ms)~n", [TotalTime/(1000*TotalCount)]),
    io:format("TOTAL count: ~p~n", [TotalCount]),
    {TotalTime, TotalCount}.

local_db_test() ->
    Keys = list_keys(),
    local_db_test(Keys).

local_db_test(Keys) ->
    {SumTime, Count} = 
    lists:foldr(fun(Key, {Sum,Count}) ->
                    {Time, [{gis_data, Key, _Cell}]} = 
                        timer:tc(mnesia, dirty_read, [gis_data, Key]),
                    {Sum + Time, Count+1}
                end, 
                {0,0},
                Keys),
    io:format("average: ~p(microsecond)~n", [SumTime/length(Keys)]),
    io:format("count: ~p~n", [Count]).
    
plocal_db_test(N) ->
    Keys = list_keys(),
    plocal_db_test(Keys, N).
plocal_db_test(Keys, N) ->
    paral_local_read(N, Keys, {0,0}).

paral_local_read(N, Keys, {AccCount, AccTime}) when N =< length(Keys) ->
    Collector = self(),
    {Keys2, Rest} = lists:split(N, Keys),
    lists:foreach(
    fun(Key) ->
        spawn(fun() ->
            {Time, [{gis_data, Key, _Cell}]} = timer:tc(mnesia, dirty_read, [gis_data, Key]),
            Collector ! {tc, Time}
        end)
    end, Keys2),
    Count = length(Keys2),
    SumTime = collect_result(Count, 0),
    io:format("parallel local average: ~p(microsecond), count=~p~n", [SumTime/Count, Count]),
    paral_local_read(N, Rest, {AccCount+Count, AccTime+SumTime});
paral_local_read(_N, Keys, {AccCount, AccTime})->
    Collector = self(),
    lists:foreach(
    fun(Key) ->
        spawn(fun() ->
            {Time, [{gis_data, Key, _Cell}]} = timer:tc(mnesia, dirty_read, [gis_data, Key]),
            Collector ! {tc, Time}
        end)
    end, Keys),
    Count = length(Keys),
    SumTime = collect_result(Count, 0),
    TotalCount = AccCount + Count,
    TotalTime = AccTime + SumTime,
    io:format("TOTAL parallel local average: ~p(microsecond)~n", [TotalTime/TotalCount]),
    io:format("TOTAL count: ~p~n", [TotalCount]).


collect_result(0, AccTime) ->
    AccTime;
collect_result(N, AccTime) ->
    receive
        {tc, Time} ->
            collect_result(N-1, AccTime+Time)
    end.

list_keys() ->
    ListFile = "./list18.txt",
    {ok, IoDevice} = file:open(ListFile, read),
    io:get_line(IoDevice, ''),
    Keys = do_loop(IoDevice, []),
    Keys.

do_loop(IoDevice, Acc) ->
    case io:get_line(IoDevice, '') of
        eof ->
            Acc;
        Line ->
            [Id | Rest] = string:tokens(Line, ".: \n"),
            <<Key:128>> = erlang:md5(Id),
            do_loop(IoDevice, [Key|Acc])
    end.